package com.dlink.sqljobsync.listener;

import com.dlink.sqljobsync.event.KafkaNoticeEvent;
import com.dlink.sqljobsync.propertie.JobConfig;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.configurationprocessor.json.JSONException;
import org.springframework.context.ApplicationEvent;
import org.springframework.context.event.EventListener;
import org.springframework.http.MediaType;
import org.springframework.stereotype.Component;
import org.springframework.web.client.RestTemplate;
import org.springframework.web.reactive.function.client.WebClient;
import reactor.core.publisher.Mono;

import javax.annotation.Resource;

/**
 * 定义所有监听事件
 * @author ASUS
 */
@Component
@Slf4j
public class AllEventListener {
    private WebClient webClient = WebClient.builder().baseUrl("http://192.168.1.4:9999/openapi/executeSql").build();
    private RestTemplate template =new RestTemplate();
    private static ObjectMapper mapper = new ObjectMapper();
    /**
     * 监听事件
     * @param event
     */
    @SneakyThrows
    @EventListener
    public void listenEvent(ApplicationEvent event) {
        /**
         * 在这里处理kafka Sql 脚本
         */
        if(event instanceof KafkaNoticeEvent){
            KafkaNoticeEvent KafkaNoticeEvent = (KafkaNoticeEvent)event;
            log.info("事件监听器获取到 KafkaNoticeEvent ，睡眠当前线程2s");
            try {
                execKafkaNoticeEvent(KafkaNoticeEvent);
                log.info("KafkaNoticeEvent 的Sql属性是: " + KafkaNoticeEvent.getSql());
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 执行KafKa事件调用
     */
    private void execKafkaNoticeEvent(KafkaNoticeEvent KafkaNoticeEvent) throws JSONException, JsonProcessingException {
        JobConfig jobConfig = new JobConfig();
        jobConfig.setAddress(KafkaNoticeEvent.getFlinkAddress());
        jobConfig.setType(KafkaNoticeEvent.getFlinkysType());
        jobConfig.setStatement(KafkaNoticeEvent.getSql());
        jobConfig.setJobName("Kafka_deom_测试");
        String json = mapper.writeValueAsString(jobConfig);
        log.info("打印请求Dlinky Sql Api json请求：");
        log.info(json);
        // 发送请求
        Mono mono = webClient.post() // POST 请求
                    .uri("http://192.168.1.4:9999/openapi/executeSql") // 请求路径
                    .contentType(MediaType.APPLICATION_JSON_UTF8)
                    .syncBody(jobConfig)
                    .retrieve() // 获取响应体
                    .bodyToMono(String.class); //响应数据类型转换
        System.out.println(mono.block());
    }
}
